
#ifndef __BLOCKING_QUEUE_H__
#define __BLOCKING_QUEUE_H__

#include <mutex>
#include <vector>
#include <memory>
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <limits>
#include <utility>


#define BLOCKING_QUEUE_IMPL_WITH_STORAGE 1


#if BLOCKING_QUEUE_IMPL_WITH_STORAGE
#include <type_traits>
#endif

template <typename T>
class BlockingQueue
{
public:
    static std::shared_ptr<BlockingQueue<T>> create(size_t max_size = 1024)
    {
        if (max_size == 0) {
            return nullptr;
        }
        return std::shared_ptr<BlockingQueue<T>>(new BlockingQueue<T>(max_size));
    }

    bool push(const T& value, size_t timeout_ms = 0)
    {
        return push_impl(value, timeout_ms);
    }

    bool push(T&& value, size_t timeout_ms = 0)
    {
        return push_impl(std::move(value), timeout_ms);
    }

    bool pop(T& value, size_t timeout_ms = 0)
    {
        if (stopped_) {
            return false;
        }

        std::unique_lock<std::mutex> lock(mutex_);
        if (timeout_ms > 0 && count_ == 0) {
            auto deadline = std::chrono::steady_clock::now() + 
                           std::chrono::milliseconds(timeout_ms);
            while (count_ == 0 && !stopped_) {
                if (not_empty_.wait_until(lock, deadline) == std::cv_status::timeout) {
                    return false;
                }
            }
        }
        if (stopped_ || count_ == 0) {
            return false;
        }
#if BLOCKING_QUEUE_IMPL_WITH_STORAGE
        T* element_ptr = reinterpret_cast<T*>(&buffer_[head_]);
        value.~T();
        new (&value) T(std::move(*element_ptr));
        element_ptr->~T();
#else
        value = std::move(buffer_[head_]);
#endif
        head_ = (head_ + 1) % max_size_;
        count_--;
        lock.unlock();
        not_full_.notify_one();

        return true;
    }

    void stop() {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            stopped_ = true;
        }
        not_empty_.notify_all();
        not_full_.notify_all();
    }

    size_t size() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return count_;
    }

    bool empty() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return count_ == 0;
    }

    bool is_stopped() const {
        return stopped_.load();
    }

    size_t capacity() const {
        return max_size_;
    }

private:
    template<typename U>
    bool push_impl(U&& value, size_t timeout_ms) {
        if (stopped_) {
            return false;
        }
        std::unique_lock<std::mutex> lock(mutex_);
        if (timeout_ms > 0 && count_ >= max_size_) {
            auto deadline = std::chrono::steady_clock::now() + 
                           std::chrono::milliseconds(timeout_ms);
            while (count_ >= max_size_ && !stopped_) {
                if (not_full_.wait_until(lock, deadline) == std::cv_status::timeout) {
                    return false;
                }
            }
        }
        if (count_ >= max_size_ || stopped_) {
            return false;
        }
#if BLOCKING_QUEUE_IMPL_WITH_STORAGE
        new (&buffer_[tail_]) T(std::forward<U>(value));
#else
        buffer_[tail_] = std::forward<U>(value);
#endif
        tail_ = (tail_ + 1) % max_size_;
        count_ ++;
        lock.unlock();
        not_empty_.notify_one();
        return true;
    }

    BlockingQueue(size_t max_size):
        max_size_(max_size), buffer_(max_size),
        head_(0), tail_(0), count_(0), stopped_(false)
    {
    }

private:
#if BLOCKING_QUEUE_IMPL_WITH_STORAGE
    std::vector<std::aligned_storage_t<sizeof(T), alignof(T)>> buffer_;
#else
    std::vector<T> buffer_;
#endif
    size_t head_;
    size_t tail_;
    size_t count_;
    mutable std::mutex mutex_;
    std::condition_variable not_empty_;
    std::condition_variable not_full_;
    std::atomic<bool> stopped_;
    size_t max_size_;
};


#endif /* __BLOCK_QUEUE_H__ */
